#include "CompressThread.h"

void CompressThread::ImageHandle::operator()()
{
    while (!exitFlag)
    {
        {
            std::unique_lock lck(tlck);
            inCond.wait(lck,[this]{return (rawSize!=0 && taskReady) || exitFlag;});
            if(exitFlag) return;
            cmpSize=LZ4_compress_fast_continue(lz4Stream_body,(char*)doubleBuff[buffPoint%2],(char*)resBuff,rawSize,maxCmpSize,Acceleration);
            resReady=true;
            taskReady=false;
        }
        outCond.notify_one();
    }
}

void CompressThread::ImageHandle::InitHandle(long size,int acceleration)
{
    std::lock_guard lck1(tlck);
    Acceleration=acceleration;
    rawSize=size;
    buffPoint=0;
    maxCmpSize=LZ4_COMPRESSBOUND(size);
    if(lz4Stream_body!=nullptr)
        delete[] lz4Stream_body;
    lz4Stream_body=new LZ4_stream_t();
    if(doubleBuff[0]!=nullptr)
    {
        delete[] doubleBuff[0];
        delete[] doubleBuff[1];
    }
    doubleBuff[0]=new unsigned char[size];
    doubleBuff[1]=new unsigned char[size];
    if(resBuff!=nullptr)
        delete[] resBuff;
    resBuff=new unsigned char[maxCmpSize];
}


void CompressThread::ImageHandle::setTask(unsigned char* data,int64_t timeSp)
{
    {   
        outlck.lock();
        std::lock_guard lck(tlck);
        buffPoint++;
        memcpy(doubleBuff[buffPoint%2],data,rawSize);
        taskReady=true;
        timesp=timeSp;
    }
    inCond.notify_one();
}

void CompressThread::ImageHandle::getRes(unsigned char* &data,int32_t &size,int64_t& timeSp)
{
    std::unique_lock lck(tlck);
    outCond.wait(lck,[this]{return exitFlag.load() || resReady.load();});
    if(exitFlag.load())
    {
      lck.unlock();
      outlck.unlock();
      throw std::exception();
    }
    data=resBuff;
    size=cmpSize;
    resBuff=new unsigned char[maxCmpSize];
    resReady=false;
    timeSp=timesp;
    outlck.unlock();
    lck.unlock();
}

void CompressThread::ImageHandle::exitThread()
{
    {
        std::lock_guard lck(tlck);
        exitFlag.store(true);
    }
    inCond.notify_all();
    if(runThread.joinable())
        runThread.join();
}

CompressThread::ImageHandle::~ImageHandle()
{
    {
        std::lock_guard lck(tlck);
        exitFlag.store(true);
    }
    inCond.notify_all();
    outCond.notify_all();
    if(runThread.joinable())
        runThread.join();
    if(lz4Stream_body!=nullptr)
        delete[] lz4Stream_body;
}

CompressThread::ImageHandle::ImageHandle()
{
    rawSize=0;
    runThread=std::thread(&CompressThread::ImageHandle::operator(),this);
}

CompressThread::CompressThread(int poolSize,long size,int acceleration) :
    threadPool(poolSize),ThreadPoolSize(poolSize)
{
    for(int i=0;i<poolSize;i++)
        threadPool[i].InitHandle(size,acceleration);
}

CompressThread::~CompressThread()
{
    exitFlag=true;
}

void CompressThread::PutImageTask(unsigned char* data,int64_t time)
{
    if(exitFlag) return;
    threadPool[putPos].setTask(data,time);
    putPos=(putPos+1)%ThreadPoolSize;
    std::cout<<"New Image Get: "<<InCount++<<std::endl;
}

void CompressThread::GetImageRes(unsigned char* &data,int32_t &size,int64_t &time)
{
    threadPool[getPos].getRes(data,size,time);
    getPos=(getPos+1)%ThreadPoolSize;
    std::cout<<"Handle Image Put: "<<OutCount++<<std::endl;
}
